/****************************************************************************
 *   Copyright (C) 2008 by 7552                                             *
 *   fiubataller2-08c1@gruposyahoo.com.ar                                   *
 *                                                                          *
 *  Licensed under the Apache License, Version 2.0 (the "License");         *
 *  you may not use this file except in compliance with the License.        *
 *  You may obtain a copy of the License at                                 *
 *                                                                          *
 *      http://www.apache.org/licenses/LICENSE-2.0                          *
 *                                                                          *
 *  Unless required by applicable law or agreed to in writing, software     *
 *  distributed under the License is distributed on an "AS IS" BASIS,       *
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*
 *  See the License for the specific language governing permissions and     *
 *  limitations under the License.                                          *
 *                                                                          *
 ***************************************************************************/
#include "RecoveryManager.h"
#include <iostream>
using namespace std;
using namespace Taller2::TransRec;

//seteo inicial de insatncia
RecoveryManager * RecoveryManager::instance = NULL;

typedef map<int,int> CheckpointMap;

//**********************************************************
RecoveryManager::RecoveryManager()
{
}

//**********************************************************
RecoveryManager* RecoveryManager::Instance(IScheduler* pScheduler)
{
	if ( !RecoveryManager::instance)
	{
	        RecoveryManager::instance = new RecoveryManager();
		instance->ope=pScheduler;
	}
    return RecoveryManager::instance;
}

//**********************************************************
int  RecoveryManager::Recover(const string& database)
{
	int tid,result;
	LogEntryType entryType;
	LogEntryType entryType_ant;
	map<int,streampos>::iterator iter;

	streampos pos;
	streampos pos_ant;

	//iniciacion del proceso
	Init(database);

	//recorrido hacia atras
	pos= logCache->GetLastEntryType(entryType);

	if(pos<0) //error de lectura en log
	{
		End();
		return result;
	}


	while(pos!=(streampos)BEGIN_OF_FILE && entryType!=CHECKPOINT )
	{

		logCache->GetTransactionID(pos, tid);

		if (tid==-1) //error de lectura del log
		{
				End();
				return ERR_FILE_READING;
		}
		if(entryType==INSERT || entryType==UPDATE || entryType==DELETE)
		{
			//busco en MapaCommited
			iter= MapaCommit.find(tid);
			if (iter==MapaCommit.end())
			{
				result=Rollback(pos,entryType);//si no hizo commit debo deshacerla
				if(!IsSuccess(result))
				{
					End();
					return result;
				}

			}
		}

		if(entryType==COMMIT )
		{
			MapaCommit[tid]=pos; //se hizo un commit
		}

		pos_ant=pos;
		entryType_ant=entryType;
		pos=logCache->GetPrevEntryType(pos, entryType); //lee proximo tipo
		if(pos==(streampos)ERR_FILE_READING)
		{
			End();
			return result;
		}
	}

	if (entryType==CHECKPOINT )
	//deshace las transacciones que no hicieron comit
	{

		CCheckPointRecord chpnt; //registro de checkpoint
		CheckpointMap trMap;     //mapa de transacciones activas del checkpoint
		CheckpointMap::iterator it; //iterador de trMap

		result=logCache->LoadReg(pos,chpnt,true);

		if(!IsSuccess(result))
		{
			End();
			return result;
		}
		trMap=chpnt.GetTransactionMap();
		
		//recorre el mapa del checkpoint hacia atras
		it = trMap.end();
		do 
		{
			it--;

			tid=it->first;
			iter= MapaCommit.find(tid);
			//si es una transaccion que no hizo commit
			if (iter==MapaCommit.end())
			{
				result=RollbackTrans(it->second);	//deshace toda la transaccion 
				if(!IsSuccess(result))
				{
					End();
					return result;
				}
			}
		}
		while (it!=trMap.begin());

	}


	//recorro hacia adelante

	entryType=entryType_ant;
	//si es el primer registro
	if (pos== (streampos)BEGIN_OF_FILE)
		pos=0;
    streampos pos_n;
	while(pos>=0 )
	{

		pos_n=logCache->GetNextEntryType(pos, entryType);

		if (pos_n==(streampos)ERR_FILE_READING)
			break;
		logCache->GetTransactionID(pos, tid);

		if(entryType==INSERT || entryType==UPDATE || entryType==DELETE)
		{
			iter= MapaCommit.find(tid);
			if (iter!=MapaCommit.end())
			{
				result=Redo(pos,entryType);//si hizo commit la rehago
				if(!IsSuccess(result))
				{
					End();
					return result;
				}
			}
		}
		pos=pos_n;
	}

	//finalizacion del proceso
	End();
	return NO_ERROR;

}

//**********************************************************
int  RecoveryManager::Redo(streampos lsn, LogEntryType entryType)
{
	int result;
	if( entryType==INSERT)
	{
		CInsertRecord IReg;
		result=logCache->LoadReg(lsn,IReg,true);
		if(!IsSuccess(result))
		{
			return result;
		}
		ObjectId oid;
		ope->createObject(IReg.Object(),oid);
	}
	if( entryType==DELETE)
	{
		CDeleteRecord DReg;
		result=logCache->LoadReg(lsn,DReg,false);
		if(!IsSuccess(result))
		{
			return result;
		}
		ObjectId oid;
		oid.instanceId = DReg.ObjectID();
		ope->deleteObject(oid);
	}

	if( entryType==UPDATE)
	{
		CUpdateRecord UReg;
		result=logCache->LoadReg(lsn,UReg,true);
		if(!IsSuccess(result))
		{
			return result;
		}
		ope->updateObject(UReg.NewObject());
	}
	return NO_ERROR;

}

//**********************************************************
int  RecoveryManager::RollbackTrans(streampos lsn)
{

	CBasicOperationRecord *Reg;
	LogEntryType entryType;
	int result;

	//lee el tipo de registro	
	lsn=logCache->GetPrevEntryType(lsn+streampos(sizeof(int)),entryType);
	
	if (lsn<0)
	{
		return ERR_ROLLBACK;
	}

	//crea un registro del mismo tipo
	if( entryType==UPDATE) Reg=new CUpdateRecord;
	else if( entryType==DELETE) Reg=new CDeleteRecord;
	else if( entryType==INSERT) Reg=new CInsertRecord;
	else return ERR_ROLLBACK; // no es un basic operation*/
	

	//lo carga desde el log
	result=logCache->LoadReg(lsn,*Reg,true);
	if(!IsSuccess(result))
		{
			delete Reg;		
			return result;
		}

	bool exito=true;

	//mientras no sea el primero de la transaccion
	while(exito)
	{
		result=Rollback(lsn,entryType); //deshace la operacion
		if(!IsSuccess(result))
		{
			delete Reg;		
			return result;
		}

		exito=logCache->GetPrevRecordOfSameTransaction(Reg); //busca el registro anterior de la misma trans.
		lsn=Reg->GetLSN();
		entryType=Reg->EntryType();
	}
	delete Reg;
	return NO_ERROR;
}

//**********************************************************
int  RecoveryManager::Rollback(streampos lsn, LogEntryType entryType)
{
	int result;

	if( entryType==INSERT)//si hizo insert hago delete
	{
		CInsertRecord IReg;
		result=logCache->LoadReg(lsn,IReg,false);
		if(!IsSuccess(result))
		{
			return result;
		}
		ObjectId oid;
		oid.instanceId = IReg.ObjectID();
		ope->deleteObject(oid);
	}
	if( entryType==DELETE)//si hizo delete hago insert
	{
		CDeleteRecord DReg;
		result=logCache->LoadReg(lsn,DReg,true);
		if(!IsSuccess(result))
		{
			return result;
		}

		ObjectId oid;
		ope->createObject(DReg.Object(),oid);
	}

	if( entryType==UPDATE)//si actualizo, vuelvo al valor anterior
	{
		CUpdateRecord UReg;
		result=	logCache->LoadReg(lsn,UReg,true);
		if(!IsSuccess(result))
		{
			return result;
		}
		ope->updateObject(UReg.OldObject());
	}

	return NO_ERROR;

}

//**********************************************************
void  RecoveryManager::Init(const string& database)
{
	logCache=new LogCache(database);
}

//**********************************************************
void  RecoveryManager::End()
{
	delete logCache;
}
